package com.wuwangfu.func;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.wuwangfu.entity.LogInfo;
import com.wuwangfu.utils.GeoUtils;
import org.apache.commons.httpclient.HttpClient;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import redis.clients.jedis.Jedis;

import java.util.HashMap;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-07-03
 */
public class ToJsonMapFunc extends RichMapFunction<String, LogInfo> {

    private transient Jedis jedis = null;
    private transient HttpClient httpClient = null;


    @Override
    public void open(Configuration parameters) throws Exception {
        //获取一个HTTP连接、Redis的连接
        ParameterTool params = (ParameterTool)getRuntimeContext()
                .getExecutionConfig()
                .getGlobalJobParameters();//获取全局参数

        String host = params.getRequired("redis.host");
        String password = params.getRequired("redis.password");
        int db = params.getInt("redis.db", 0);
        jedis = new Jedis(host, 6379, 5000);
        jedis.auth(password);
        jedis.select(db);
        //访问高德地图API
        httpClient = new HttpClient();
    }

    @Override
    public LogInfo map(String line) throws Exception {
        //判断jedis是否断开，断开重连
        if (!jedis.isConnected()){
            jedis.connect();
        }
        LogInfo logInfo = null;

        try {
            String[] fields = line.split("-->");
            String dateTime = fields[0];
            String dt = dateTime.split(" ")[0];
            String json = fields[1];
            JSONObject jsonObj = JSON.parseObject(json);

            JSONObject uObj = jsonObj.getJSONObject("u");
            JSONObject phoneObj = uObj.getJSONObject("phone");
            JSONObject locObj = uObj.getJSONObject("loc");
            JSONObject appObj = uObj.getJSONObject("app");

            //取出user对象中的扁平字段
            String id = uObj.getString("id");
            String account = uObj.getString("account");
            String sessionId = uObj.getString("sessionId");

            //取出手机设备信息
            String imei = phoneObj.getString("imei");
            String osName = phoneObj.getString("osName");
            String osVer = phoneObj.getString("osVer");
            String resolution = phoneObj.getString("resolution");
            String androidId = phoneObj.getString("androidId");
            String manufacture = phoneObj.getString("manufacture");
            String deviceId = phoneObj.getString("deviceId");

            //取出loc位置信息
            String areacode = locObj.getString("areacode");
            double longtitude = locObj.getDouble("longtitude");
            double latitude = locObj.getDouble("latitude");

            //根据经纬度查找省份、市、商圈
            String[] areaInfo = GeoUtils.getAreaInfoByLongitudeAndLatitude(httpClient, jedis, longtitude, latitude);

            String province = areaInfo[0];
            String city = areaInfo[1];
            String district = areaInfo[2];
            String bizNames = areaInfo[3];

            String carrier = locObj.getString("carrier");
            String netType = locObj.getString("netType");
            String cid_sn = locObj.getString("cid_sn");
            String ip = locObj.getString("ip");

            //取出app各个字段
            String appid = appObj.getString("appid");
            String appVer = appObj.getString("appVer");
            String release_ch = appObj.getString("release_ch");
            String promotion_ch = appObj.getString("promotion_ch");

            //事件类型
            String logType = jsonObj.getString("logType");
            //提交时间
            Long commit_time = jsonObj.getLong("commit_time");

            JSONObject eventObj = jsonObj.getJSONObject("event");
            //构造一个用于装event数据的hashmap
            HashMap<String, String> eventMap = new HashMap<>();
            //迭代取出event中每一个kv
            for (String k : eventObj.keySet()){
                String v = eventObj.getString(k);
                //添加到hahmap中
                eventMap.put(k,v);
            }
            //组装数据并返回
            logInfo = new LogInfo(id,
                    account,
                    sessionId,
                    imei,
                    osName,
                    osVer,
                    resolution,
                    androidId,
                    manufacture,
                    deviceId,
                    areacode,
                    longtitude,
                    latitude,
                    province,
                    city,
                    district,
                    bizNames,
                    carrier,
                    netType,
                    cid_sn,
                    ip,
                    appid,
                    appVer,
                    release_ch,
                    promotion_ch,
                    logType,
                    commit_time,
                    dt,
                    eventMap
            );
        } catch (Exception e) {
            e.printStackTrace();
        }

        return logInfo;
    }

    @Override
    public void close() throws Exception {
        //关闭HTTP连接和redis连接
        jedis.close();
        httpClient = null;
    }
}